package small.common.core.file.center;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import small.common.core.file.event.FileChangeEvent;
import small.common.core.file.watcher.FileWatcher;

import java.io.File;
import java.io.IOException;
import java.nio.file.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Author zhoujin
 * @Date 2022/11/9 17:07
 */
public class WatchFileCenter {

    protected static final Logger LOGGER = LoggerFactory.getLogger(WatchFileCenter.class);

    private static final AtomicBoolean CLOSED = new AtomicBoolean(false);

    private static final int MAX_WATCH_FILE_JOB = 16;

    private static final Map<String, WatchDirJob> MANAGER = new HashMap<String, WatchDirJob>(MAX_WATCH_FILE_JOB);

    private static int NOW_WATCH_JOB_CNT = 0;

    private static final FileSystem FILE_SYSTEM = FileSystems.getDefault();

    public static synchronized boolean registerWatcher(final String paths, FileWatcher watcher){
        checkState();
        if (NOW_WATCH_JOB_CNT == MAX_WATCH_FILE_JOB) {
            return false;
        }
        WatchDirJob job = MANAGER.get(paths);
        if (job == null) {
            job = new WatchDirJob(paths);
            job.start();
            MANAGER.put(paths, job);
            NOW_WATCH_JOB_CNT++;
        }
        job.addSubscribe(watcher);
        return true;
    }

    private static void checkState() {
        if (CLOSED.get()) {
            throw new IllegalStateException("WatchFileCenter already shutdown");
        }
    }

    private static class WatchDirJob extends Thread {

        private ExecutorService callBackExecutor;

        private final String paths;

        private Map<FileWatcher,Boolean> watchers = new ConcurrentHashMap();

        private WatchService watchService;

        private volatile boolean watch = true;

        public WatchDirJob(String paths){
            setName(paths);
            this.paths = paths;
            final Path p = Paths.get(paths);
            if (!p.toFile().isDirectory()) {
                throw new IllegalArgumentException("Must be a file directory : " + paths);
            }

            this.callBackExecutor = Executors.newFixedThreadPool(1, new ThreadFactory() {
                @Override
                public Thread newThread(Runnable runnable) {
                    String threadName = paths;
                    Thread thread = new Thread(runnable, threadName);
                    thread.setDaemon(true);
                    return thread;
                }
            });

            try {
                WatchService service = FILE_SYSTEM.newWatchService();
                p.register(service, StandardWatchEventKinds.OVERFLOW, StandardWatchEventKinds.ENTRY_MODIFY,
                        StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_DELETE);
                this.watchService = service;
            } catch (IOException e) {
                e.printStackTrace();
            }

        }

        void addSubscribe(final FileWatcher watcher) {
            watchers.put(watcher,true);
        }


        @Override
        public void run() {
            while (watch) {
                try {
                    final WatchKey watchKey = watchService.take();
                    final List<WatchEvent<?>> events = watchKey.pollEvents();
                    watchKey.reset();
                    if (callBackExecutor.isShutdown()) {
                        return;
                    }
                    if (events.isEmpty()) {
                        continue;
                    }
                    callBackExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            for (WatchEvent<?> event : events) {
                                WatchEvent.Kind<?> kind = event.kind();

                                // Since the OS's event cache may be overflow, a backstop is needed
                                if (StandardWatchEventKinds.OVERFLOW.equals(kind)) {
                                    eventOverflow();
                                } else {
                                    eventProcess(event.context());
                                }
                            }
                        }
                    });
                } catch (InterruptedException ignore) {
                    Thread.interrupted();
                } catch (Throwable ex) {
                    LOGGER.error("An exception occurred during file listening : {}", ex);
                }
            }
        }

        private void eventProcess(Object context) {
            final FileChangeEvent fileChangeEvent = FileChangeEvent.builder().paths(paths).context(context).build();
            final String str = String.valueOf(context);
            for (final FileWatcher watcher : watchers.keySet()) {
                if (watcher.interest(str)) {
                    Runnable job = new Runnable() {
                        @Override
                        public void run() {
                            watcher.onChange(fileChangeEvent);
                        }
                    };
                    Executor executor = watcher.executor();
                    if (executor == null) {
                        try {
                            job.run();
                        } catch (Throwable ex) {
                            LOGGER.error("File change event callback error : {}", ex);
                        }
                    } else {
                        executor.execute(job);
                    }
                }
            }
        }

        private void eventOverflow() {
            File dir = Paths.get(paths).toFile();
            for (File file : Objects.requireNonNull(dir.listFiles())) {
                // 子目录不参与监听
                if (file.isDirectory()) {
                    continue;
                }
                eventProcess(file.getName());
            }
        }
    }

}
